package cn.tianyigps.m7.command;

import cn.tianyigps.m7.beans.Task;
import cn.tianyigps.m7.template.KafkaTemplate;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.IOException;
import java.util.Arrays;
import java.util.Map;
import java.util.Optional;

/**
 * Created with IDEA
 *
 * @description: 指令监听
 * @author: cui.guibin
 * @create: 2019/4/10 15:40
 **/

public class CommandConsumer{

    private static Logger log = LogManager.getLogger(CommandConsumer.class);

    public static void main(String[] args) {
        KafkaConsumer<String, String> consumer = KafkaTemplate.consumer();
        consumer.subscribe(Arrays.asList("m7_command"));
        while (true){
            ConsumerRecords<String, String> consumerRecords = consumer.poll(10);
            ObjectMapper objectMapper = new ObjectMapper();
            if (null != consumerRecords && consumerRecords.count() > 0){
                for (ConsumerRecord consumerRecord:consumerRecords) {
                    String value = (String) consumerRecord.value();
                    Command command = null;
                    try {
                        command = objectMapper.readValue(value, Command.class);
                    } catch (IOException e) {
                        log.error(e.getMessage(),"指令下发格式错误 + " + value);
                    }
                    if (!Optional.ofNullable(command).isPresent()) {
                        log.error("指令为空");
                        return;
                    }
                    try {
                        Command.sendCommand(command);
                    } catch (IOException e) {
                        Command.instructionReply(command.getEquipmentId(),e.getMessage(),406);
                    }
                }
            }
        }
    }
}
